package com.sunday.common.leaf.lock;

import com.sunday.common.core.gson.GsonUtils;
import com.sunday.common.core.utils.DateTimeUtils;
import com.sunday.common.core.utils.NetUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * 利用zookeeper实现HA锁
 * 参考资料 :
 * https://zookeeper.apache.org/releases.html
 * https://blog.csdn.net/qq_37960603/article/details/121835169
 * https://blog.csdn.net/aihao2006/article/details/117299941
 *
 * @author zsy
 * @since 2022/10/25
 */
@Slf4j
public class HALock {

    private static final String PATH_PREFIX = "/lock";
    private static final String PATH_SUFFIX = "/high";

    private String address;

    private String path;
    /**
     * 等待节点
     */
    private String waitNode;
    /**
     * 当前节点
     */
    private String currNode;
    /**
     * 当前运行状态
     */
    private volatile boolean running;

    private volatile CountDownLatch latch;

    private CuratorFramework curator;

    private HALock(String address, String group, String classify) {
        this.address = address;
        this.path = PATH_PREFIX + "/" + group + PATH_SUFFIX + "/" + classify;
        log.info("[HA] address : {}, path : {}", address, path);
        init();
    }

    private void init() {
        try {
            curator = createWithOptions(address, new RetryUntilElapsed(1000, 3), 10000, 6000);
            curator.start();
            // 监控根路径下节点变化
            watcher(curator);
            // 创建临时有序节点
            createEphSeqNode();
        } catch (Exception e) {
            log.error("[HA] init error : {}", e.toString(), e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 监控节点情况
     *
     * @param curator
     */
    private void watcher(CuratorFramework curator) {
        CuratorCache cache = CuratorCache.build(curator, path);
        CuratorCacheListener listener = CuratorCacheListener.builder().forPathChildrenCache(path, curator, (client, event) -> {
            // 节点删除 / 连接丢失 均进行放锁核验处理
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED || event.getType() == PathChildrenCacheEvent.Type.CONNECTION_LOST) {
                log.info("[HA] event type : {}, node election", event.getType());
                unlock();
            }
        }).build();
        cache.listenable().addListener(listener);
        cache.start();
    }

    /**
     * 获得锁, 未获得锁则等待
     */
    public void lock() {
        try {
            // 未运行 + 未命中 = 等待
            while (!running && notHit()) {
                await();
            }
        } catch (Exception e) {
            log.error("[HA] lock error : {}", e.toString(), e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 释放锁
     */
    private void unlock() {
        log.info("[HA] owner : {} , current node : {} ", running, currNode);
        if (!running && this.latch != null) {
            this.latch.countDown();
        }
    }

    private boolean notHit() throws Exception {
        return !hit();
    }

    /**
     * 是否命中锁，并更新等待临时有序节点信息
     *
     * @return {@link boolean}
     */
    private boolean hit() throws Exception {
        // 获取全部子节点并排序
        List<String> children = getChildrenSort();
        // 验证自己是否为最小节点
        if (currNode.equals(path + "/" + children.getFirst())) {
            log.info("[HA] current node : {} , has been locked", currNode);
            //启动运行锁
            return (running = true);
        }
        // 非最小节点获取前一个节点信息
        String currKey = currNode.substring(currNode.lastIndexOf("/") + 1);
        int pointer = Collections.binarySearch(children, currKey);
        // 检测自己的节点是否丢失
        if (pointer < 0) {
            // 丢失进行创建并复测
            createEphSeqNode();
            return hit();
        } else {
            // 获取前一个节点信息
            waitNode = children.get(pointer - 1);
            return false;
        }
    }

    /**
     * 等待锁发生变动
     */
    private void await() throws Exception {
        // 检测当前临时有序节点是否存在
        if (curator.checkExists().forPath(path + "/" + waitNode) != null) {
            this.latch = new CountDownLatch(1);
            this.latch.await();
            this.latch = null;
        }
    }

    /**
     * 获取根目录下所有子节点并排序
     *
     * @return {@link List< String>}
     */
    private List<String> getChildrenSort() throws Exception {
        //获取全部子节点
        List<String> children = curator.getChildren().forPath(path);
        //如果集合为空则，重新设置当前临时节点，并重新获取集合
        while (children.isEmpty()) {
            createEphSeqNode();
            children = curator.getChildren().forPath(path);
        }
        return children.stream().sorted().collect(Collectors.toList());
    }

    /**
     * 创建临时有序节点
     */
    private void createEphSeqNode() throws Exception {
        currNode = curator.create()
                .creatingParentContainersIfNeeded()
//                .withProtection()
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath(path + "/eph-seq-", buildData().getBytes());
        log.info("[HA] initial node : {}", currNode);
    }


    private CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy, int connectionTimeoutMs, int sessionTimeoutMs) {
        return CuratorFrameworkFactory.builder().connectString(connectionString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
    }

    /**
     * 端点信息
     *
     * @author zsy
     * @since 2022/10/26
     */
    public record Endpoint(String ip, String timestamp) {
    }

    private String buildData() {
        return GsonUtils.SERIALIZE_NULL.toJson(new Endpoint(NetUtils.getLocalAddress().getHostAddress(), DateTimeUtils.formatDateTimeMsSplit()));
    }

    public static HALock of(String address, String directory, String group) {
        return new HALock(address, directory, group);
    }

}
